package kafka.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.message.HotArticleConstants;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CountNumProducer {

    public static final String TOPIC = "count-num-stream-topic";
    private static final String KAFKA_IP = "192.168.179.100:9092";
    public static void main(String[] args) throws InterruptedException {


        //添加kafka的配置信息
        Properties properties = new Properties();
        //配置broker信息
        properties.put("bootstrap.servers",KAFKA_IP);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.RETRIES_CONFIG,10);

        //生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);

        for(int i=0; i<10;i++){
            ArticleBehaviorMsg viewMsg = new ArticleBehaviorMsg();
            viewMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS);
            viewMsg.setArticleId(1664913291440345090L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(viewMsg)));

            ArticleBehaviorMsg likeMsg = new ArticleBehaviorMsg();
            likeMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.LIKES);
            likeMsg.setArticleId(1664913291440345090L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(likeMsg)));

            ArticleBehaviorMsg collectMsg = new ArticleBehaviorMsg();
            collectMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.COLLECTION);
            collectMsg.setArticleId(1664913291440345090L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(collectMsg)));

            ArticleBehaviorMsg commentMsg = new ArticleBehaviorMsg();
            commentMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.COMMENT);
            commentMsg.setArticleId(1664913291440345090L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(commentMsg)));


            ArticleBehaviorMsg viewMsg2 = new ArticleBehaviorMsg();
            viewMsg2.setType(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS);
            viewMsg2.setArticleId(1664913186800848897L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(viewMsg2)));
            System.out.println("消息发送完成"+(i+1));
            Thread.sleep(2000);
        }
        System.out.println("消息发送完成");

/*        for (int i=1;i<100;i++) {
            
            //发送消息
            try {
                String msg1 = "a1:2";
                ProducerRecord<String,String> record1 = new ProducerRecord<String, String>(TOPIC,msg1);
                producer.send(record1);

                String msg2 = "a2:1";
                ProducerRecord<String,String> record2 = new ProducerRecord<String, String>(TOPIC,msg2);
                producer.send(record2);
            }catch (Exception e){
                e.printStackTrace();
            }
           
            Thread.sleep(2000);
        }*/
        //关系消息通道
        producer.close();
    }
}
